package streamAPI.operators;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.testng.annotations.Test;
import util.model.OrderBean;
import util.model.WordCount;
import util.FlinkConstant;


/**
 * @author yue.cao
 * @since 09-28-2020
 */
public class DemoWithKeyBy {

	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStreamSource<String> source = env.fromElements(FlinkConstant._Common.SOME_STRING);
		source.map(new MapFunction<String, WordCount>() {
			@Override
			public WordCount map(String value) throws Exception {
				return WordCount.of(value, 1L);
			}
		}).keyBy("word").sum("count").print();
		env.execute();
	}

	@Test
	public void t1() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStreamSource<String> source = env.fromElements("上海,上海,100", "山东,青岛,100", "上海,上海,100", "美国,阿富汗,1", "山东,青岛,100");
		source.map(new MapFunction<String, OrderBean>() {
			@Override
			public OrderBean map(String value) throws Exception {
				String[] split = value.split(",");
				return OrderBean.of(split[0], split[1], Double.parseDouble(split[2]));
			}
		}).keyBy("province", "city").sum("money").print();
		env.execute();
	}

	@Test(description = "reduce")
	public void t2() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStreamSource<String> source = env.fromElements(FlinkConstant._Common.SOME_STRING);
		source.map(new MapFunction<String, WordCount>() {
			@Override
			public WordCount map(String value) throws Exception {
				return WordCount.of(value, 1L);
			}
		}).keyBy("word").reduce(new ReduceFunction<WordCount>() {
			@Override
			public WordCount reduce(WordCount v1, WordCount v2) throws Exception {
				return WordCount.of(v1.word, v1.count + v2.count);
			}
		}).print();
		env.execute();
	}

	@Test(description = "max")
	public void t3() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStreamSource<String> source = env.fromElements("a,1","a,2","a,3","a,4","b,5","b,1");
		source.map(new MapFunction<String, Tuple2<String,Integer>>() {
			@Override
			public Tuple2<String, Integer> map(String line) throws Exception {
				String[] split = line.split(",");
				return Tuple2.of(split[0],Integer.parseInt(split[1]));
			}
		}).keyBy(0).max(1).print();
		env.execute();
	}



}
